"""
Case Type   : 行存压缩
Case Name   : 普通/分区/二级分区表，主机修改压缩参数，主机循环进行增删改查操作后，备机增量重建,主备数据一致
Create At   : 2023/8/11
Owner       : li-xin12345
Description :
    1、设置同步参数
    2、建普通非压缩表，并插入数据，修改为压缩表
    3、建分区压缩表，并插入数据，修改压缩参数
    4、建二级分区压缩表，并插入数据，修改为非压缩表
    5、主机三表循环执行增删改查操作时，备机stop，主机继续插入完成
    6、备机重建，查看主备数据一致
    7、清理环境
Expect      :
    1、成功
    2、成功
    3、成功
    4、成功
    5、成功
    6、一致
    7、成功
History     :
    Modified by li-xin12345 at 2023/11/6:环境差异，sql线程未执行完成，调整sql循环次数及线程等待时间
    Modified by li-xin12345 at 2023/11/23:备机重建后，增加等待时间，再校验数据库状态
    Modified by li-xin12345 at 2025/3/24:增加重启数据库操作，若可恢复集群，避免影响后续用例
"""

import os
import unittest
import re
import time

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread
from testcase.utils.Logger import Logger

Pri_U_SH = CommonSH('PrimaryDbUser')
Node_Num = Pri_U_SH.get_node_num()


@unittest.skipIf(3 > Node_Num, '不满足一主两备环境跳过')
class RowCompressTest(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'-----{os.path.basename(__file__)} start-----')
        self.std1_sh = CommonSH('Standby1DbUser')
        self.std2_sh = CommonSH('Standby2DbUser')
        self.constant = Constant()
        self.common = Common()
        self.tb_name = ['t_alter_rowcompress_0082_01',
                        't_alter_rowcompress_0082_02',
                        't_alter_rowcompress_0082_03']
        self.create_compress_param = ['compresstype=1', 'compresstype = 2']
        self.alter_compress_param = ['compresstype=2, '
                                     'compress_byte_convert=true, '
                                     'compress_diff_convert=true, '
                                     'compress_level=16, '
                                     'compress_chunk_size=2048, '
                                     'compress_prealloc_chunks=3',
                                     'compresstype=2, '
                                     'compress_byte_convert=true, '
                                     'compress_diff_convert=true, '
                                     'compress_level=31, '
                                     'compress_chunk_size=2048, '
                                     'compress_prealloc_chunks=2',
                                     'compresstype=0']
        self.guc_params = ['synchronous_commit', 'most_available_sync ']
        self.default_params = [self.common.show_param(f'{i}') for i in
                               self.guc_params]
        self.log.info(self.default_params)

    def test_row_compress(self):
        text = '-----step1:设置同步参数;expect:成功-----'
        self.log.info(text)
        for i in self.guc_params:
            res = Pri_U_SH.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG, f'{i}=on')
            self.log.info(res)
            self.assertTrue(res)

        text = '-----step2:建普通非压缩表，并插入数据，修改为压缩表;expect:成功-----'
        self.log.info(text)
        sql = f'''drop table if exists {self.tb_name[0]} cascade;
        create table {self.tb_name[0]}(columnone integer, 
        columntwo char(50), columnthree varchar(50), columnfour integer, 
        columnfive char(50), columnsix varchar(50), columnseven char(50), 
        columneight char(50), columnnine varchar(50), columnten varchar(50), 
        columneleven char(50), columntwelve char(50), 
        columnthirteen varchar(50), columnfourteen char(50), 
        columnfifteem varchar(50));
        insert into {self.tb_name[0]} values(generate_series(0, 1000), 
        'test',substring(md5(random()::text), 1, 16), 
        2, substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16));
        '''
        sql_res = Pri_U_SH.execut_db_sql(sql)
        self.log.info(sql_res)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, sql_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, sql_res,
                      '执行失败：' + text)
        self.assertIn('INSERT 0 1001', sql_res, '执行失败：' + text)

        text = '---修改为压缩表---'
        self.log.info(text)
        sql = f'alter table {self.tb_name[0]} set ' \
              f'({self.alter_compress_param[0]}); checkpoint;' \
              f'\\d+ {self.tb_name[0]}'
        sql_res = Pri_U_SH.execut_db_sql(sql)
        self.log.info(sql_res)
        self.assertIn(self.constant.ALTER_TABLE_MSG, sql_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.CHECKPOINT_SUCCESS_MSG, sql_res,
                      '执行失败：' + text)
        self.assertIn(f'{self.alter_compress_param[0]}', sql_res,
                      '执行失败：' + text)

        text = '-----step3:建分区压缩表，并插入数据，修改压缩参数;expect:成功-----'
        self.log.info(text)
        sql = f'''drop table if exists {self.tb_name[1]} cascade;
        create table {self.tb_name[1]}
        (prod_id number(6),
         cust_id number,
         time_id date,
         channel_id char(1),
         promo_id number(6),
         quantity_sold number(3),
         amount_sold number(10,2)
        )
        with ({self.create_compress_param[0]})
        partition by range (time_id)
        interval('1 day')
        ( partition p1 values less than ('2023-02-01 00:00:00'),
          partition p2 values less than ('2023-02-02 00:00:00')
        );
        insert into {self.tb_name[1]} values(generate_series(0, 1000), 
        lpad(floor(random() * 1000)::text, 2, '0'), 
        '2023-01-10 00:00:00', 'a', 1, 1, 1);
        insert into {self.tb_name[1]} values(generate_series(0, 1000), 
        lpad(floor(random() * 1000)::text, 2, '0'), 
        '2023-02-01 00:00:00', 'a', 2, 2, 2);
        '''
        sql_res = Pri_U_SH.execut_db_sql(sql)
        self.log.info(sql_res)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, sql_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, sql_res,
                      '执行失败：' + text)
        self.assertIn('INSERT 0 1001', sql_res, '执行失败：' + text)
        self.assertEqual(sql_res.count('INSERT 0 1001'), 2, '执行失败：' + text)

        text = '---修改压缩参数---'
        self.log.info(text)
        sql = f'alter table {self.tb_name[1]} set ' \
              f'({self.alter_compress_param[1]}); checkpoint;' \
              f'\\d+ {self.tb_name[1]}'
        sql_res = Pri_U_SH.execut_db_sql(sql)
        self.log.info(sql_res)
        self.assertIn(self.constant.ALTER_TABLE_MSG, sql_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.CHECKPOINT_SUCCESS_MSG, sql_res,
                      '执行失败：' + text)
        self.assertIn(f'{self.alter_compress_param[1]}', sql_res,
                      '执行失败：' + text)

        text = '-----step4:建二级分区压缩表，并插入数据，修改为非压缩表;' \
               'expect:成功-----'
        self.log.info(text)
        sql = f'''drop table if exists {self.tb_name[2]} cascade;
        create table {self.tb_name[2]}
        (
            month_code varchar2 ( 30 ) not null ,
            dept_code  varchar2 ( 30 ) not null ,
            user_no    varchar2 ( 30 ) not null ,
            sales_amt  int
        )
        with ({self.create_compress_param[1]})
        partition by list (month_code) subpartition by list (dept_code)
        (
          partition p_201901 values ( '201902' )
          (
            subpartition p_201901_a values ( '1' ),
            subpartition p_201901_b values ( '2' )
          ),
          partition p_201902 values ( '201903' )
          (
            subpartition p_201902_a values ( '1' ),
            subpartition p_201902_b values ( '2' )
          )
        );
        insert into {self.tb_name[2]} values('201902', '1', '1', 1);
        insert into {self.tb_name[2]} values('201902', '2', '1', 1);
        insert into {self.tb_name[2]} values('201902', '1', '1', 1);
        insert into {self.tb_name[2]} values('201903', '2', '1', 1);
        insert into {self.tb_name[2]} values('201903', '1', '1', 1);
        insert into {self.tb_name[2]} values('201903', '2', '1', 1);
        '''
        sql_res = Pri_U_SH.execut_db_sql(sql)
        self.log.info(sql_res)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, sql_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.TABLE_CREATE_SUCCESS, sql_res,
                      '执行失败：' + text)
        self.assertIn('INSERT 0 1', sql_res, '执行失败：' + text)
        self.assertEqual(sql_res.count('INSERT 0 1'), 6, '执行失败：' + text)

        text = '---改为非压缩表-----'
        self.log.info(text)
        sql = f'alter table {self.tb_name[2]} set ' \
              f'({self.alter_compress_param[2]}); checkpoint;' \
              f'\\d+ {self.tb_name[2]}'
        sql_res = Pri_U_SH.execut_db_sql(sql)
        self.log.info(sql_res)
        self.assertIn(self.constant.ALTER_TABLE_MSG, sql_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.CHECKPOINT_SUCCESS_MSG, sql_res,
                      '执行失败：' + text)
        self.assertIn(f'{self.alter_compress_param[2]}', sql_res,
                      '执行失败：' + text)

        text = '-----step5:主机三表循环执行增删改查操作时，备机stop，' \
               '主机继续插入完成;expect:成功-----'
        self.log.info(text)
        sql = f'''update {self.tb_name[0]} set columnone = 9999 
        where columnone <= 500;
        update {self.tb_name[1]} set prod_id = 9999 where prod_id <= 500;
        update {self.tb_name[2]} set sales_amt = 33 where dept_code = '1';
        insert into {self.tb_name[0]} values(generate_series(0, 1000), 
        'test',substring(md5(random()::text), 1, 16), 
        2, substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16), 
        substring(md5(random()::text), 1, 16));
        insert into {self.tb_name[1]} values(generate_series(0, 1000), 
        lpad(floor(random() * 1000)::text, 2, '0'), 
        '2023-01-10 00:00:00', 'a', 1, 1, 1);
        insert into {self.tb_name[1]} values(generate_series(0, 1000), 
        lpad(floor(random() * 1000)::text, 2, '0'), 
        '2023-02-01 00:00:00', 'a', 2, 2, 2);
        insert into {self.tb_name[2]} values('201902', '1', '1', 1);
        insert into {self.tb_name[2]} values('201902', '2', '1', 1);
        insert into {self.tb_name[2]} values('201902', '1', '1', 1);
        insert into {self.tb_name[2]} values('201903', '2', '1', 1);
        insert into {self.tb_name[2]} values('201903', '1', '1', 1);
        insert into {self.tb_name[2]} values('201903', '2', '1', 1);
        delete from {self.tb_name[0]} where columnone = (
        select columnone from {self.tb_name[0]} order by random() limit 1);
        delete from {self.tb_name[1]} where prod_id = (
        select prod_id from {self.tb_name[1]} order by random() limit 1);
        delete from {self.tb_name[2]} where sales_amt = (
        select sales_amt from {self.tb_name[2]} order by random() limit 1);
        select count(*) from {self.tb_name[0]};
        select count(*) from {self.tb_name[1]};
        select count(*) from {self.tb_name[2]};
        '''
        dml_thread = ComThread(Pri_U_SH.cycle_exec_sql,
                               args=(sql, 100))
        dml_thread.setDaemon(True)
        dml_thread.start()
        time.sleep(10)

        self.log.info('---备机stop---')
        std1_res = self.std1_sh.stop_db_instance()
        std2_res = self.std2_sh.stop_db_instance()
        self.log.info(std1_res)
        self.log.info(std2_res)
        self.assertIn(self.constant.GS_CTL_STOP_SUCCESS_MSG, std1_res,
                      '执行失败：' + text)
        self.assertIn(self.constant.GS_CTL_STOP_SUCCESS_MSG, std2_res,
                      '执行失败：' + text)

        dml_thread.join(120)
        dml_res = dml_thread.get_result()
        self.log.info(dml_res)
        self.assertTrue(dml_res)

        text = '-----备机重建，查看主备数据一致;expect:成功，一致-----'
        self.log.info(text)
        build_res1 = self.std1_sh.build_standby(mode='-b incremental')
        self.assertIn(self.constant.BUILD_SUCCESS_MSG, build_res1)
        build_res2 = self.std2_sh.build_standby(mode='-b incremental')
        self.assertIn(self.constant.BUILD_SUCCESS_MSG, build_res2)
        time.sleep(30)
        status = Pri_U_SH.get_db_cluster_status(param='status')
        self.log.info(status)
        self.assertTrue(status)

        self.log.info('---查看数据一致---')
        for i in range(3):
            sql = f'select checksum({self.tb_name[i]}::text) ' \
                  f'from {self.tb_name[i]};'
            pri_res = Pri_U_SH.execut_db_sql(sql)
            self.log.info(pri_res)
            std1_res = self.std1_sh.execut_db_sql(sql)
            self.log.info(std1_res)
            std2_res = self.std2_sh.execut_db_sql(sql)
            self.log.info(std2_res)
            self.assertEqual(pri_res, std1_res, '执行失败：' + text)
            self.assertEqual(pri_res, std2_res, '执行失败：' + text)
            self.assertEqual(std1_res, std2_res, '执行失败：' + text)

    def tearDown(self):
        text = '-----step7:清理环境; expect:成功-----'
        self.log.info(text)
        restart = Pri_U_SH.restart_db_cluster()
        self.log.info(restart)
        status = Pri_U_SH.get_db_cluster_status(param='status')
        self.log.info(status)
        guc_params = [self.common.show_param(f'{i}') for i in self.guc_params]
        restore_guc = []
        for i in range(2):
            if guc_params[i] != self.guc_params[i]:
                restore_guc.append(Pri_U_SH.execute_gsguc(
                    'reload', self.constant.GSGUC_SUCCESS_MSG,
                    f'{self.guc_params[i]}={self.default_params[i]}'))
                self.log.info(restore_guc)

        drop_res = []
        for k, v in enumerate(self.tb_name):
            sql = f'drop table if exists {v};'
            drop_res.append(Pri_U_SH.execut_db_sql(sql))
            self.log.info(drop_res)
            self.assertIn(self.constant.DROP_TABLE_SUCCESS, drop_res[k],
                          '执行失败：' + text)
            k += 1

        [self.assertTrue(i) for i in restore_guc]
        self.assertEqual(
            sum(1 for s in drop_res if re.search(r'DROP\s+TABLE', s)), 3,
            '执行失败：' + text)
        self.log.info(f'-----{os.path.basename(__file__)} end-----')